package com.googlecode.jvmpart.support;

import java.util.Collection;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.googlecode.jvmpart.Partitioner;
import com.googlecode.jvmpart.ProcessExecution;
import com.googlecode.jvmpart.ProcessExecutionSplitter;
import com.googlecode.jvmpart.ProcessListener;
import com.googlecode.jvmpart.ProcessManager;
import com.googlecode.jvmpart.ProcessPartitionHandler;
import com.googlecode.jvmpart.proc.JvmProcess;
import com.googlecode.jvmpart.proc.JvmProcessException;

/**
 * Using the partitioner with a process spliter {@link ProcessExecutionSplitter}
 * to split a set of the {@link ProcessExecution}
 * for starting the process monitor threads which execute 
 * and monitor the JVM processes parallel
 * then queue the JVM processes when the capacity of the parallel is full
 * @author chunzhan
 *
 */
public class JvmProcessPatitionHandler implements ProcessPartitionHandler,ProcessListener{
	private static final Log logger = LogFactory.getLog(JvmProcessPatitionHandler.class);
	private ProcessListener listener;
	private String jvmProcessClass;
	private int capacity;
	/**
	 * @param jvmProcessClass the canonical class name of the JVM process, 
	 * which impletements the interface {@link JvmProcess}
	 * @param capacity the size of the parallel JVM processes
	 */
	public JvmProcessPatitionHandler(String jvmProcessClass, int capacity) {
		this.jvmProcessClass = jvmProcessClass;
		this.capacity = capacity;
	}
	/**
	 * @param jvmProcessClass the class of the JVM process, 
	 * which impletements the interface {@link JvmProcess}
	 * @param capacity the size of the parallel JVM processes
	 */
	public JvmProcessPatitionHandler(Class<? extends JvmProcess> jvmProcessClass, int capacity) {
		this(jvmProcessClass.getCanonicalName(), capacity);
	}
	/**
	 * there are not capacity for the JVM processes parralel
	 * All processes will be executed almost at the same time.
	 * @param jvmProcessClass the class of the JVM process, 
	 * which impletements the interface {@link JvmProcess}
	 */
	public JvmProcessPatitionHandler(Class<? extends JvmProcess> jvmProcessClass) {
		this(jvmProcessClass.getCanonicalName(), -1);
	}
	public Collection<ProcessExecution> handle(ProcessExecutionSplitter jvmSplitter,
			Partitioner partitioner, ProcessManager processManager) throws JvmProcessException {
		Set<ProcessExecution> set = jvmSplitter.split(partitioner, capacity);
		int total = set.size();
		CountDownLatch countDownLatch = new CountDownLatch(total);
		int startCount = 0;
		for(ProcessExecution threadExecution:set) {
			//queue when the capacity is full
			while(capacity>0
					&&startCount>=capacity
					&&(countDownLatch.getCount()+(startCount-capacity))==total) {
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					logger.error("queue the JVM process", e);
				}
			}
			logger.debug("start process:"+threadExecution.getName());
			threadExecution.setJvmProcessClass(jvmProcessClass);
			if(processManager!=null) {
				StringBuffer buf = new StringBuffer();
				buf.append("-D");
				buf.append(ProcessManager.class.getCanonicalName());
				buf.append("=");
				buf.append(processManager.getClass().getCanonicalName());
				threadExecution.addVmArgument(buf.toString());
			}
			threadExecution.setCountDownLatch(countDownLatch);
			threadExecution.setProcessListener(this);
			threadExecution.start();
			startCount++;
		}
		try {
			countDownLatch.await();
		} catch (InterruptedException e) {
			logger.error(this, e);
		} finally {
			if(processManager!=null)
				processManager.exit(ProcessExecution.PID, ProcessExecution.PID_TIMESTAMP, set);
		}
		return set;
	}
	/**
	 * Using a sample JVM process splitter to handle JVM processes
	 * and persistent the processes status locally
	 * @param partitioner
	 * @return a collection of thread context for aggregate
	 */
	public Collection<ProcessExecution> handle(Partitioner partitioner) throws JvmProcessException{
		return handle(new SampleJvmExecutionSpliter(), partitioner, new LocalProcessManager());
	}
	/**
	 * Using a sample JVM process splitter to handle JVM processes
	 * @param partitioner
	 * @param processManager manage the process status persistent
	 * @return a collection of thread context for aggregate
	 */
	public Collection<ProcessExecution> handle(Partitioner partitioner, ProcessManager processManager) throws JvmProcessException{
		return handle(new SampleJvmExecutionSpliter(), partitioner, processManager);
	}
	/**
	 * @param listener JVM process listener
	 */
	public void setProcessListener(ProcessListener listener) {
		this.listener = listener;
	}
	public synchronized void beforeProcess(ProcessExecution threadExecution) {
		if(listener!=null) {
			listener.beforeProcess(threadExecution);
		}		
		
	}
	public synchronized void afterProcess(ProcessExecution threadExecution) {
		if(listener!=null) {
			listener.afterProcess(threadExecution);
		}		
	}
}
